package com.liu.spike.controller;

import com.alibaba.fastjson.JSONObject;
import com.liu.spike.common.KafkaTopic;
import com.liu.spike.entity.Product;
import com.liu.spike.entity.dto.ProductOrderDto;
import com.liu.spike.service.KafkaProducerServer;
import lombok.RequiredArgsConstructor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.Collections;

/**
 * @className: com.liu.spike.controller.SplikeController
 * @description: TODO
 * @author: liu
 * @create: 2022-05-01 23:34
 */

@RequiredArgsConstructor
@RequestMapping("/splike")
@RestController
public class SplikeController {

    @Autowired
    DefaultRedisScript spikeScript;
//    https://www.jianshu.com/p/a555facfd6c8
    @Autowired
    private RedisTemplate<String,Object> redisTemplate;

    @Autowired
    private KafkaProducerServer kafkaProducerServer;

    @PostMapping("/do")
    public String doIsOk(@RequestBody ProductOrderDto dto){
        String productKeyState = getProductKeyState(dto.getProductId());
        String state = (String) redisTemplate.opsForValue().get(productKeyState);
        if(state.equals("end")){
            return "秒杀结束";
        }
        String productKey = getProductKey(dto.getProductId());
        Long decrement = redisTemplate.opsForValue().decrement(productKey);
        if(decrement >= 0){
            if(decrement == 0){
                redisTemplate.opsForValue().set(productKeyState,"end");
            }
            kafkaProducerServer.sendToTopic("splikeTopic", JSONObject.toJSONString(dto));
        }
        return null;
    }


    @PostMapping("/doIt")
    public String doIsOk2(@RequestBody ProductOrderDto dto){
        Long result = (Long) redisTemplate.execute(spikeScript, Collections.singletonList(dto.getProductId()+""),dto.getUid()+"");
        if(result == 1L){
            kafkaProducerServer.sendToTopic(KafkaTopic.SPIKE_TOPIC, JSONObject.toJSONString(dto));
            return "秒杀成功";
        }
        return result == 0 ? "秒杀成功":"秒杀失败";
    }


    private String getProductKey(Integer product){
        return "product"+"-"+product;
    }

    private String getProductKeyState(Integer product){
        return "product"+"-"+"state"+"-"+product;
    }
}
